package com.yamed.bus.service.kakfa;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

/**
 * @Author: wchen
 * @Description:
 * @Date: Created in 2019-08-19 11:50
 * Copyright 本内容仅限于杭州阿拉丁信息科技股份有限公司内部传阅，禁止外泄以及用于其他的商业目的
 */
public class ProducerApi {
    private final Logger logger = LoggerFactory.getLogger(this.getClass());
    private Producer producer;

    ProducerApi(){
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.111.34:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        this.producer = new KafkaProducer<>(props);
    }

    public Producer<String, String> getProducer() {
        return producer;
    }


    /**
     * 同步阻塞发送
     */
    public void syncHandle() {
        try {
//            byte[] key = "key".getBytes();
//            byte[] value = "value".getBytes();
//            ProducerRecord<byte[], byte[]> record = new ProducerRecord<byte[], byte[]>("my-topic", key, value);
//            producer.send(record).get();
            String key = "key";
            String value = "value";
            ProducerRecord<String, String> record = new ProducerRecord<String, String>("my-topic", key, value);
            producer.send(record).get();
        }catch (Exception e){
            logger.error("syncHandle error", e);
        }
    }

    /**
     * 异步发送(无阻塞通知)
     */
    public void asyncHandle(){
        String key = "key";
        String value = "value";
        ProducerRecord<String, String> record = new ProducerRecord<String, String>("the-topic", key, value);
        producer.send(record,
            new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception e) {
                    logger.info("asyncHandle.Callback ...");
                    if(e != null) {
                        e.printStackTrace();
                    }
                    System.out.println("The offset of the record we just sent is: " + metadata.offset());
                }
            });
    }

    /**
     * 生产者测试
     */
    public void producerTestData(){
        Producer producer = getProducer();
        for (int i = 0; i < 4; i++) {
            producer.send(new ProducerRecord<>("my-topic", Integer.toString(i), Integer.toString(i)));
        }
        producer.close();
    }

    public static void main(String[] args) throws Exception{
        ProducerApi producerApi = new ProducerApi();
        producerApi.producerTestData();
//        producerApi.syncHandle();
//        producerApi.asyncHandle();

//        Thread.sleep(1000*10);
    }
}
